package com.jthinking.service.kafka.delay.queue.listener;

import com.jthinking.service.kafka.delay.queue.storage.StorageService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;


@Component
@ConditionalOnProperty(value = {"storage.enable"})
public class TaskListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(TaskListener.class);

    @Resource
    private StorageService storageService;

    @KafkaListener(
            topics = {"delay-queue-task"},
            properties = {
                    "key.deserializer:org.apache.kafka.common.serialization.LongDeserializer",
                    "value.deserializer:org.apache.kafka.common.serialization.StringDeserializer",
            }
    )
    public void listen(List<ConsumerRecord<Long, String>> list) {
        long start = System.nanoTime();

        for (ConsumerRecord<Long, String> record : list) {
            Long second = record.key();
            storageService.add(second, record.value().getBytes());
            LOGGER.info("storageService.add second: {}, record.value: {}", second, record.value());
        }

        LOGGER.info("TaskListener.listen size: {}, nano time: {}", list.size(), System.nanoTime() - start);
    }

}
